package io.github.bigbird0101.easycluster.client;

import io.github.bigbird0101.easycluster.URL;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

public class NettyClient extends AbstractClient{
    static String NETTY_EPOLL_ENABLE_KEY = "netty.epoll.enable";
    static String OS_NAME_KEY = "os.name";
    static String OS_LINUX_PREFIX = "linux";
    int DEFAULT_CONNECT_TIMEOUT = 3000;
    int DEFAULT_HEARTBEAT = 60 * 1000;
    int DEFAULT_IO_THREADS = Math.min(Runtime.getRuntime().availableProcessors() + 1, 32);
    private Bootstrap bootstrap;

    private final Lock connectLock =new ReentrantLock();

    private Channel channel;

    public NettyClient(URL url) {
        super(url);
    }

    public void start(){
        bootstrap = new Bootstrap();
        bootstrap.channel(createChannel());
        bootstrap.group(createEventLoopGroup());
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.max(DEFAULT_CONNECT_TIMEOUT, getConnectTimeout()));
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                ch.pipeline()
                        .addLast("decoder",new StringDecoder())
                        .addLast("encoder",new StringEncoder())
                        .addLast("client-idle-handler",
                                new IdleStateHandler(DEFAULT_HEARTBEAT, 0, 0, MILLISECONDS))
                        .addLast("handler", new SimpleChannelInboundHandler<Object>(){
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
                                System.out.println(msg);
                                send(msg);
                            }
                        });
            }
        });
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(()->{
            try{

                if(null==this.channel || !this.channel.isActive()){
                    System.out.println("重连中。。。");
                    reconnect();
                }
            }catch (Throwable ignored){

            }
        },2000L,getConnectTimeout(), MILLISECONDS);
    }


    public void connect(){
        connectLock.lock();
        try {
            if(null!=this.channel && this.channel.isActive()){
                System.out.println("连接是成功的状态");
                return;
            }
            ChannelFuture future = bootstrap.connect(new InetSocketAddress(getUrl().getHost(), getUrl().getPort()));
            boolean ret = future.awaitUninterruptibly(getConnectTimeout(), MILLISECONDS);
            if (ret && future.isSuccess()) {
                Channel newChannel = future.channel();
                try {
                    if (null != this.channel) {
                        this.channel.close();
                    }
                } finally {
                   this.channel = newChannel;
                }
                System.out.println("连接成功");
            } else if (future.cause() != null) {
                System.out.println("连接失败 " + future.cause().getCause().getMessage());
            } else {
                System.out.println("连接失败");
            }
        }finally {
            connectLock.unlock();
        }
    }


    private int getConnectTimeout() {
        return (int) getUrl().getAttribute().getOrDefault("DEFAULT_CONNECT_TIMEOUT",DEFAULT_CONNECT_TIMEOUT);
    }

    private EventLoopGroup createEventLoopGroup() {
        ThreadFactory threadFactory=new DefaultThreadFactory("NettyClient",true);
        return shouldEpoll()?new EpollEventLoopGroup(DEFAULT_IO_THREADS,threadFactory):new NioEventLoopGroup(DEFAULT_IO_THREADS,threadFactory);
    }

    private Class<? extends Channel> createChannel() {
        return shouldEpoll()? EpollSocketChannel.class : NioSocketChannel.class;
    }

    private static boolean shouldEpoll() {
        if (Boolean.parseBoolean(System.getProperty(NETTY_EPOLL_ENABLE_KEY, "false"))) {
            String osName = System.getProperty(OS_NAME_KEY);
            return osName.toLowerCase().contains(OS_LINUX_PREFIX) && Epoll.isAvailable();
        }
        return false;
    }

    @Override
    public void send(Object message) {
        this.channel.writeAndFlush(message);
    }

    @Override
    public void reconnect() {
        if(null==this.channel || !this.channel.isActive()) {
            connect();
        }
    }
}
